package org.xiaoyu.rpc.remoting.transport.cllient.future;

import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.xiaoyu.rpc.remoting.packet.AppResponse;
import org.xiaoyu.rpc.remoting.packet.Invocation;
import org.xiaoyu.rpc.remoting.packet.PacketStatusConstant;

import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Created By Have
 * 2021/7/16 15:41
 */
@Slf4j
public class RpcFuture extends CompletableFuture<Object> {
    private static final Map<String, RpcFuture> FUTURES = new ConcurrentHashMap<>();
    private static final Map<String, Channel> CHANNELS = new ConcurrentHashMap<>();

    public static final Timer TIME_OUT_TIMER = new HashedWheelTimer(
      new NamedThreadFactory("rpc-timout-future", true),
            30,
            TimeUnit.MILLISECONDS
    );

    private final Channel channel;
    private final Invocation invocation;
    private final long start = System.currentTimeMillis();
    private volatile long sent;
    private final int timeout;
    private final String id;
    private Timeout timeoutCheckTask;


    private RpcFuture(Channel channel, Invocation invocation, int timeout) {
        this.channel = channel;
        this.invocation = invocation;
        this.timeout = timeout;
        this.id = invocation.getRequestId();
        FUTURES.put(this.id, this);
        CHANNELS.put(this.id, channel);
    }

    private static void timeoutCheck(RpcFuture future) {
        TimeoutCheckTask task = new TimeoutCheckTask(future.getId());
        future.timeoutCheckTask = TIME_OUT_TIMER.newTimeout(task, future.getTimeout(), TimeUnit.MILLISECONDS);
    }

    public static RpcFuture newFuture(Channel channel, Invocation invocation, int timeout) {
        RpcFuture rpcFuture = new RpcFuture(channel, invocation, timeout);
        timeoutCheck(rpcFuture);
        return rpcFuture;
    }

    public static void sent(Channel channel, Invocation invocation) {
        RpcFuture future = FUTURES.get(invocation.getRequestId());
        if (future != null) {
            future.doSent();
        }
    }

    public static void received(Channel channel, AppResponse appResponse) {
        received(channel, appResponse, false);
    }

    public static void received(Channel channel, AppResponse response, boolean isTimeout) {
        try {
            RpcFuture remove = FUTURES.remove(response.getRequestId());
            if (remove != null) {
                Timeout t = remove.timeoutCheckTask;
                if (!isTimeout) {
                    t.cancel();
                }
                remove.doReceived(response);
            } else {
                log.warn("The timeout response finally returned at "
                        + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
                        + ", response status is " + response.getStatusCode()
                        + (channel == null ? "" : ", channel: " + channel.localAddress()
                        + " -> " + channel.remoteAddress()) + ", please check provider side for detailed result.");
            }
        } finally {
            CHANNELS.remove(response.getRequestId());
        }
    }

    private void doReceived(AppResponse response) {
        if (response == null) {
            throw new IllegalStateException("response cannot be null");
        }
        if (response.getStatusCode() == PacketStatusConstant.SUCCESS) {
            this.complete(response);
        } else if (response.getStatusCode() == PacketStatusConstant.TIMEOUT) {
            this.completeExceptionally(new RuntimeException(response.getRequestId() + " is Timeout," + response.getErrorMessage()));
        } else {
            this.completeExceptionally(new RuntimeException(response.getErrorMessage()));
        }
    }

    public static RpcFuture getFuture(String requestId) {
        return FUTURES.get(requestId);
    }

    public static boolean hasFuture(Channel channel) {
        return CHANNELS.containsValue(channel);
    }

    private void doSent() {
        sent = System.currentTimeMillis();
    }

    private boolean isSent() {
        return sent > 0;
    }

    public String getId() {
        return id;
    }

    public int getTimeout() {
        return timeout;
    }

    public Channel getChannel() {
        return channel;
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        AppResponse errorResult = new AppResponse();
        errorResult.setRequestId(id);
        errorResult.setErrorMessage("request future has been canceled");
        errorResult.setStatusCode(PacketStatusConstant.ERROR);
        this.doReceived(errorResult);
        FUTURES.remove(id);
        CHANNELS.remove(id);
        return true;
    }

    private String getTimeoutMessage(boolean scan) {
        long nowTimestamp = System.currentTimeMillis();
        return (sent > 0 ? "Waiting server-side response timeout" : "Sending request timeout in client-side")
                + (scan ? " by scan timer" : "") + ". start time: "
                + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date(start))) + ", end time: "
                + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date(nowTimestamp))) + ","
                + (sent > 0 ? " client elapsed: " + (sent - start)
                + " ms, server elapsed: " + (nowTimestamp - sent)
                : " elapsed: " + (nowTimestamp - start)) + " ms, timeout: "
                + timeout + " ms, request: " + (log.isDebugEnabled() ? invocation : getRequestWithoutData()) + ", channel: " + (InetSocketAddress) channel.localAddress()
                + " -> " + (InetSocketAddress) channel.remoteAddress();
    }

    private String getRequestWithoutData() {
        invocation.setParameters(null);
        return invocation.toString();
    }

    private static class TimeoutCheckTask implements TimerTask {
        private final String requestID;


        TimeoutCheckTask(String requestID) {
            this.requestID = requestID;
        }

        @Override
        public void run(Timeout timeout) {
            RpcFuture future = RpcFuture.getFuture(this.requestID);
            if (future == null || future.isDone()) {
                return;
            }

            notifyTimeout(future);
        }

        private void notifyTimeout(RpcFuture future) {
            AppResponse timeoutResponse = new AppResponse();
            timeoutResponse.setRequestId(requestID);
            timeoutResponse.setStatusCode(PacketStatusConstant.TIMEOUT);
            timeoutResponse.setErrorMessage(future.getTimeoutMessage(true));
            RpcFuture.received(future.getChannel(), timeoutResponse, true);
        }
    }


}
